package org.code.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.code.common.Constant;
import org.code.util.ByteStringConverter;
import org.code.util.ThreadUtil;
import org.junit.Test;

import java.util.List;

/**
 * @author xiao
 * <p>
 * 快速入门案例-普通消息发送接收
 * <p>
 * 发送消息和接收消息
 */
public class Simple {

    /**
     * topic
     */
    public static final String TOPIC = "simple-topic";

    /**
     * 测试生产者
     * <p>
     * 消息生产者流程：
     * 1.创建消息生产者producer，并制定生产者组名
     * 2.指定Nameserver地址
     * 3.启动producer
     * 4.创建消息对象，指定主题Topic、Tag和消息体等
     * 5.发送消息
     * 6.关闭生产者producer
     */
    @Test
    public void simpleProducer() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        // 创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("test-group");
        // 设置nameServer地址
        producer.setNamesrvAddr(Constant.NAME_SERVER);
        // 启动生产者
        producer.start();
        // 发送消息
        for (int i = 0; i < 10; i++) {
            // 创建消息对象
            /*
             * @param topic 主题,表示要发送的消息的主题
             * @param body 消息体,表示消息的存储内容
             * @param tags 标签,消息标签，用来进一步区分某个 Topic 下的消息分类，消息从生产者发出即带上的属性。
             * @param keys 消息键,Apache RocketMQ 每个消息可以在业务层面的设置唯一标识码 keys 字段，方便将来定位消息丢失问题。
             */
            Message message = new Message(TOPIC, ByteStringConverter.stringToByteArray("Hello RocketMQ" + i));
            // 发送消息
            SendResult send = producer.send(message);
            System.out.println(send);
        }
        producer.shutdown();
    }

    /**
     * 测试消费者
     * <p>
     * 消息消费者流程：
     * 1.创建消费者consumer，制定消费者组名
     * 2.指定Nameserver地址
     * 3.创建监听订阅主题Topic和Tag等
     * 4.处理消息
     * 5.启动消费者consumer
     */
    @Test
    public void simpleConsumer() throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer-group");
        // 设置nameServer地址
        consumer.setNamesrvAddr(Constant.NAME_SERVER);
        // 订阅主题
        consumer.subscribe(TOPIC, "*");
        /*
         * 监听消息
         * MessageListenerConcurrently 对象用于同时接收异步传递的消息
         * MessageListenerOrderly 对象用于有序地接收消息。一个队列一个线程
         */
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            /**
             * 消费消息
             * @param list 消息列表
             * @param consumeConcurrentlyContext 消息上下文
             * @return 消费状态
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 处理消息
                for (MessageExt messageExt : list) {
                    System.out.printf("threadName: %s, msgId : %s, queueId: %s, body: %s\n",
                            ThreadUtil.getCurrentThreadName(),
                            messageExt.getQueueId(),
                            messageExt.getMsgId(),
                            ByteStringConverter.byteArrayToString(messageExt.getBody()));
                }
                /*
                 * 消费成功
                 * @param CONSUME_SUCCESS:消费成功
                 * @param RECONSUME_LATER:消费失败，需要重试
                 */
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        // 阻塞
        System.in.read();
    }
}
